% prj2server.erl
%
% Initial function that starts the entire gossip program
%
-module(prj2).
-export([init/4, init/5]).
-import(erlang, [display/1]).

% init function
init(NumServers, FileDir, Replication, Machines) ->
    init(NumServers, FileDir, Replication, Machines, 0).
 
init(NumServers, FileDir, Replication, Machines, NumNeighbors) ->

	random:seed(now()),

	% only 2 VM for now, can add more if needed, gets IP address
	Server1 = list_to_atom(lists:nth(1, string:tokens(Machines, ","))),
	Server2 = list_to_atom(lists:nth(2, string:tokens(Machines, ","))),
        io:format("~p ~p ~n", [Server1, Server2]),

    % given number of initial processes to launch, splits it evenly between the two VMs
    Proc1 = {prj2server, Server1},
    Proc2 = {prj2server, Server2},
    Half = round(NumServers / 2),
    Proc1 ! {start, Half, 0, self()},
    Proc2 ! {start, NumServers, Half, self()},
  
	Servers = getMessage() ++ getMessage(),
    
    if
        NumNeighbors == 0 ->
            io:format("Creating neighbor list from matrix~n"),
            db:createNeighborList(1, Servers);
        true ->
            io:format("Creating random neighbor list with ~p neighbors~n", [NumNeighbors]),
            distributeNeighbors(Servers, Servers, NumNeighbors)
    end,
	
    try
		FileList = getFilesFromDirectory(FileDir),
		sendFilesToServers(FileList, Servers, Replication),
		io:format("Done adding neighbors.~n"),

		io:format("/************* Start Update Fragment ****************/~n"),
		io:format("Checking initial nodes in order to list which of them have the fragment~n"),
		checkServerFragments(Servers, 2),
		wait(5),
		io:format("Updating the nodes that have the fragment~n"),
		updateFragment(Servers, 2),
		io:format("/************* End Update Fragment - Check server terminal for output ****************/~n~n"),
		
		wait(20),
		resetBool(Servers),

		io:format("/************* Retrieve Fragment ****************/~n"),
		io:format("Retrieving fragment 1~n"),
		retrieveFragment(Servers, 1),
    io:format("/************* End Retrieve Fragment - Check server terminal for output ****************/~n~n"),

		io:format("/************* Gossip for Median ****************/~n"),
		startMedianGossiping(Servers),
		getMedian(Servers),
		io:format("/************* End Median Gossip - Check server terminal for output ****************/~n~n"),
		
		io:format("/************* Gossip for Average ****************/~n"),
		startAverageGossiping(Servers),
		wait(1000),
		io:format("/************* End Average Gossip - Check server terminal for output ****************/~n~n"),
		
		wait(10),
		
		io:format("/************* Gossip for Min ****************/~n"),
		io:format("Current Global Min:\t~p~n", [db:getMin(Servers)]),
		io:format("Start gossip for min~n"),
		getGossipMin(Servers), 
		io:format("/************* End Min Gossip - Check server terminal for output ****************/~n~n"),

		io:format("/************* Gossip for Max ****************/~n"),
    io:format("Start gossip for max~n"),
		getGossipMax(Servers), 
    io:format("/************* End Max Gossip - Check server terminal for output ****************/~n~n"),

    io:format("Stopping servers~n"),
		db:stopServers(Servers)
	catch
		Error ->
			display(Error),
			db:stopServers(Servers)
	end,
    ok.

%Max gossip
printMAX([]) ->
	wait(100),
	ok;
printMAX(Servers) ->
	hd(Servers) ! {printMax},
	printMAX(tl(Servers)).

getGossipMax([]) ->
	wait(100),
	ok;
getGossipMax(Servers) ->
	hd(Servers) ! {gmax},
	getGossipMax(tl(Servers)).

%Min gossip
printMIN([]) ->
	wait(100),
	ok;
printMIN(Servers) ->
	hd(Servers) ! {printMin},
	printMIN(tl(Servers)).

getGossipMin([]) ->
	wait(100),
	ok;
getGossipMin(Servers) ->
	hd(Servers) ! {gmin},
	getGossipMin(tl(Servers)).
    
% Distribute Neighbors
distributeNeighbors([], _Servers, _NumRandom) ->
  ok;
distributeNeighbors(ToMessage, Servers, NumRandom) ->
  RandomServers = getRandomServers(Servers, NumRandom, []),
  addNeighborsToServers(hd(ToMessage), RandomServers), 
  distributeNeighbors(tl(ToMessage), Servers, NumRandom).

addNeighborsToServers(_ToMessage, []) ->
  ok;
addNeighborsToServers(ToMessage, ToAdd) ->
  ToMessage ! {addn, hd(ToAdd)},
  addNeighborsToServers(ToMessage, tl(ToAdd)).
  
getAverage([]) ->
	ok;
getAverage(Servers) ->
	hd(Servers) ! {getavg, self()},
	io:format("~p\t~p~n", [hd(Servers), getMessage()]),
	getAverage(tl(Servers)).

getMedian([]) ->
    wait(100),
    ok;
getMedian(Servers) ->
    hd(Servers) ! {getmedian, self()},
    io:format("~p\t~p~n", [hd(Servers), getMessage()]),
    getMedian(tl(Servers)).
    
startAverageGossiping([]) ->
  ok;
startAverageGossiping(Servers) ->
  hd(Servers) ! {avg, start},
  startAverageGossiping(tl(Servers)).
  
startMedianGossiping([]) ->
  wait(100),
  ok;
startMedianGossiping(Servers) ->
  hd(Servers) ! {median, start},
  startMedianGossiping(tl(Servers)).

resetBool([]) -> ok;
resetBool(Servers) ->
	hd(Servers) ! {reset},
	resetBool(tl(Servers)).

checkServerFragments([], FragN) -> ok;
checkServerFragments(Servers, FragN) ->
	hd(Servers) ! {checkfragment, FragN},
	checkServerFragments(tl(Servers), FragN).

updateFragment(Servers, FragN) ->
	File = file:open("newfile-1.txt", read),
	case File of
		{ok, IoDevice} ->
			Floats = readFloatsFromFile(IoDevice, []),
			file:close(IoDevice);
		{error, Msg} ->
			Floats = [],
			throw({error, Msg})
	end,
	hd(Servers) ! {update, FragN, Floats},
	wait(5),
	ok.
	
retrieveFragment(Servers, FragId) ->
  hd(Servers) ! {retrieve, FragId, self()},
	wait(5),
  receive
    {Tag, FloatList, ServerId} ->
      io:format("Retrieved fragment ~p containing ~p from ~p~n", 
        [Tag, FloatList, ServerId]),
      ok;
    Other ->
      io:format("Got error ~p~n", [Other]),
      {error, unknown_msg, Other}
    after 5000 ->
      io:format("Did not receive any messages~n")
  end.

wait(Sec) -> 
    receive
    after (1000 * Sec) -> ok
    end.
    
% Gets a single response from the mailbox
getMessage() ->
	receive
		Msg ->
			Msg
	end.

%% Distributes the given list of files to a random set of Servers with the given Replication
sendFilesToServers([], _Servers, _Replication) ->
    ok;
sendFilesToServers(FileList, Servers, Replication) ->
    % Get the first file in the list
    % Get random servers that will be holding this server
    ServersToHoldFile = getRandomServers(Servers, Replication, []),   
    File = file:open(hd(FileList), read),
    case File of
        {ok, IoDevice} ->
            Floats = readFloatsFromFile(IoDevice, []),
            sendFloatsToServers(getTagFromFilename(hd(FileList)), Floats, ServersToHoldFile),
			sendFilesToServers(tl(FileList), Servers, Replication),
			file:close(IoDevice);
        {error, eisdir} ->
            sendFilesToServers(tl(FileList), Servers, Replication);            
        {error, Msg} ->
            throw({error, Msg})
    end,
    ok.

% Reads all the floats from the given file until EOF
readFloatsFromFile(IoDevice, ListOfFloats) ->
    Line = file:read_line(IoDevice),
    case Line of
        {ok, Data} ->
            readFloatsFromFile(IoDevice, [list_to_float(re:replace(Data, "\\s$", "", [global, {return, list}])) | ListOfFloats ]);
        eof ->
            ListOfFloats
    end.

%% Sends a list of floats to the set of lines
sendFloatsToServers(_Tag, _Floats, []) ->
    ok;
sendFloatsToServers(Tag, Floats, Servers) ->
    db:add(hd(Servers), Tag, Floats),
    sendFloatsToServers(Tag, Floats, tl(Servers)).
    
% Gets a list of random servers   
getRandomServers(_FullServerList, Number, Add) when length(Add) == Number ->
    Add;
getRandomServers(FullServerList, Number, Add) ->
    Index = random:uniform(length(FullServerList)),
    getRandomServers(FullServerList, Number, lists:usort([ lists:nth(Index, FullServerList) | Add ])).

%% Gets a listing of files from the given directory
getFilesFromDirectory(FileDir) when is_list(FileDir) ->
    Filenames = file:list_dir(FileDir),
    if
        element(1, Filenames) /= ok ->
            throw({error, "Error getting files from directory"});
        true ->
            addDirToFiles(FileDir, lists:sort(element(2, Filenames)), [])
    end;
getFilesFromDirectory(_FileDir) ->
    throw({exit, "FileDir must be a string"}).

%% Adds the directory string to each of the files
addDirToFiles(_FileDir, [], Add) ->
    Add;
addDirToFiles(FileDir, Filenames, Add) ->
    addDirToFiles(FileDir, tl(Filenames), [ FileDir ++ "/" ++ hd(Filenames) | Add]).

%% Gets the tag integer from a filename, expected format ".*-[0-9].txt"
getTagFromFilename(Filename) ->
    list_to_integer(re:replace(re:replace(Filename, ".*-", "", [global, {return, list}]), "\.txt", "", [global, {return, list}])).
